package org.pentaho.di.trans.steps.redis.reset;

import java.util.HashSet;
import java.util.Iterator;
import java.util.Map;
import java.util.Set;

import org.pentaho.di.core.exception.KettleException;
import org.pentaho.di.core.util.StringUtil;
import org.pentaho.di.i18n.BaseMessages;
import org.pentaho.di.trans.Trans;
import org.pentaho.di.trans.TransMeta;
import org.pentaho.di.trans.step.BaseStep;
import org.pentaho.di.trans.step.StepDataInterface;
import org.pentaho.di.trans.step.StepInterface;
import org.pentaho.di.trans.step.StepMeta;
import org.pentaho.di.trans.step.StepMetaInterface;

import org.pentaho.di.trans.steps.redis.output.RedisOutputMeta;
import redis.clients.jedis.Jedis;
import redis.clients.jedis.JedisPoolConfig;
import redis.clients.jedis.JedisSentinelPool;
import redis.clients.util.Pool;

/**
 * The Redis Input step looks up value objects, from the given key names, from Redis server(s).
 */
public class RedisReset extends BaseStep implements StepInterface {
    private static Class<?> PKG = RedisResetMeta.class; // for i18n purposes, needed by Translator2!! $NON-NLS-1$

    protected RedisResetMeta meta;
    protected RedisResetData data;
    String keytype="string" ;
    String key2 = "default";
    String valuetype ;
    String tablename ;
    String idfieldname ;
    public RedisReset(StepMeta stepMeta, StepDataInterface stepDataInterface, int copyNr, TransMeta transMeta,
                      Trans trans) {
        super(stepMeta, stepDataInterface, copyNr, transMeta, trans);
    }

    private static Pool<Jedis> pool = null;

    @Override
    public boolean init(StepMetaInterface smi, StepDataInterface sdi) {
        if (super.init(smi, sdi)) {
            try {
                pool = ((RedisOutputMeta) smi).createJedisPool();
                
                return true;
            } catch (Exception e) {
                logError(BaseMessages.getString(PKG, "RedisReset.Error.ConnectError"), e);
                return false;
            }
        } else {
            return false;
        }
    }

    public boolean processRow(StepMetaInterface smi, StepDataInterface sdi) throws KettleException {
        meta = (RedisResetMeta) smi;
        data = (RedisResetData) sdi;
        String param1="*_1_*";
        String param0="*_0_*";
        if(meta.getTableName()!=null&&!StringUtil.isEmpty(meta.getTableName())){
        	tablename = environmentSubstitute(meta.getTableName());
        	param1=tablename+"_1_*";
        	param0=tablename+"_0_*";
        }
        logBasic("tablename:"+tablename);
        Jedis jedis = pool.getResource();

        Set<String> keysdel = jedis.keys(param0);
        Iterator<String> iteratordel = keysdel.iterator();
        while (iteratordel.hasNext()) {
        	String key =iteratordel.next();
        	jedis.del(key);
		}

        Set<String> keys = jedis.keys(param1);
        Iterator<String> iterator = keys.iterator();
        while (iterator.hasNext()) {
        	String key =iterator.next();
        	String newkey=key.replaceAll("_1_", "_0_");
        	jedis.set(newkey, jedis.get(key));
        	jedis.del(key);
		}
        if (checkFeedback(getLinesRead())) {
            if (log.isBasic()) {
                logBasic(BaseMessages.getString(PKG, "RedisOutput.Log.LineNumber") + getLinesRead());
            }
        }
        jedis.close();
        setOutputDone();
        return false;
    }
    
    @Override
    public void dispose(StepMetaInterface smi, StepDataInterface sdi) {
    	super.dispose(smi, sdi);
    	if(pool!=null){
    		pool.close();
    		pool.destroy();
    	}
    }
   
}
